package com.neusoft.bigdata.hbase.biz.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.neusoft.bigdata.hbase.biz.AbstractHBaseBiz;
import com.neusoft.bigdata.hbase.biz.StatisticService;
import com.neusoft.bigdata.hbase.model.MetricData;
import com.neusoft.bigdata.hbase.model.StaticMetricData;
import com.neusoft.bigdata.hbase.model.StatisticData;
import com.neusoft.bigdata.hbase.model.StatisticData.PeriodType;
import com.neusoft.bigdata.hbase.model.StatisticData.StatisticType;

/**
 * @author lihzh
 * @date 2012-11-21 下午4:32:58
 */
public class StatisticServiceHBaseImpl extends AbstractHBaseBiz implements
		StatisticService {

	private static HTable statisticTable;

	static {
		try {
			statisticTable = new HTable(conf, TABLE_NAME_STATISTIC);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void batchSavaData(List<StatisticData> datas) {
		List<Put> puts = createStatisticPuts(datas);
		try {
			statisticTable.put(puts);
			statisticTable.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public StaticMetricData findData(String fqn, long startTime, long endTime) {
		StaticMetricData staticData = new StaticMetricData();
		Map<String, MetricData> metricAvgMap= new HashMap<String, MetricData>();
		try {
			String startRow = createRowKey(fqn, PeriodType.H.toString(),
					StatisticType.AVG.toString(), startTime);
			String endRow = createRowKey(fqn, PeriodType.H.toString(),
					StatisticType.AVG.toString(), endTime);
			Iterator<Result> resultIt = getByFQN(statisticTable, startRow, endRow);
			MetricData cpuloadData = new MetricData(COLUMN_NAME_CPULOAD);
			MetricData memoryUsageData = new MetricData(COLUMN_NAME_MEMORYUSED);
			staticData.setLastDataTime(startTime);
			while (resultIt.hasNext()) {
				Result result = resultIt.next();
				KeyValue baseInfoKv = result.getColumnLatest(Bytes.toBytes(FAMILY_NAME_BASEINFO), Bytes.toBytes(COLUMN_NAME_STARTTIME));
				long time = Long.valueOf(new String(baseInfoKv.getValue()));
				KeyValue cpuloadKv = result.getColumnLatest(
						Bytes.toBytes(FAMILY_NAME_STINFO),
						Bytes.toBytes(COLUMN_NAME_CPULOAD));
				if (cpuloadKv != null && !cpuloadKv.isEmptyColumn()) {
					Double cpuloadValue = Double.valueOf(new String(cpuloadKv
							.getValue()));
					cpuloadData.addDataPoint(cpuloadValue, time);
				}
				KeyValue memoryUsageKv = result.getColumnLatest(
						Bytes.toBytes(FAMILY_NAME_STINFO),
						Bytes.toBytes(COLUMN_NAME_MEMORYUSAGE));
				if (memoryUsageKv != null && !memoryUsageKv.isEmptyColumn()) {
					Double memoryUsageValue = Double.valueOf(new String(
							memoryUsageKv.getValue()));
					memoryUsageData.addDataPoint(memoryUsageValue, time);
				}
				staticData.setLastDataTime(time);
			}
			metricAvgMap.put(COLUMN_NAME_CPULOAD, cpuloadData);
			metricAvgMap.put(COLUMN_NAME_MEMORYUSAGE, memoryUsageData);
			staticData.setMetricDataMap(metricAvgMap);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return staticData;
	}

	private List<Put> createStatisticPuts(List<StatisticData> datas) {
		List<Put> puts = new ArrayList<Put>();
		for (StatisticData statics : datas) {
			String rowkey = createRowKey(statics);
			Put put = new Put(Bytes.toBytes(rowkey));
			Map<String, Double> metricValueMap = statics.getMetricValueMap();
			Iterator<Entry<String, Double>> mapIt = metricValueMap.entrySet()
					.iterator();
			while (mapIt.hasNext()) {
				Entry<String, Double> entry = mapIt.next();
				String metircName = entry.getKey();
				Double value = entry.getValue();
				put.add(Bytes.toBytes(FAMILY_NAME_STINFO),
						Bytes.toBytes(metircName),
						Bytes.toBytes(String.valueOf(value)));
			}
			put.add(Bytes.toBytes(FAMILY_NAME_BASEINFO),
					Bytes.toBytes(COLUMN_NAME_RESOURCEID),
					Bytes.toBytes(statics.getResourceId()));
			put.add(Bytes.toBytes(FAMILY_NAME_BASEINFO),
					Bytes.toBytes(COLUMN_NAME_PERIOD_TYPE),
					Bytes.toBytes(statics.getPeriodType().toString()));
			put.add(Bytes.toBytes(FAMILY_NAME_BASEINFO),
					Bytes.toBytes(COLUMN_NAME_STATISTIC_TYPE),
					Bytes.toBytes(statics.getStatisticType().toString()));
			put.add(Bytes.toBytes(FAMILY_NAME_BASEINFO),
					Bytes.toBytes(COLUMN_NAME_STARTTIME),
					Bytes.toBytes(String.valueOf(statics.getStartTime())));
			puts.add(put);
		}
		return puts;
	}

	private String createRowKey(StatisticData statics) {
		return createRowKey(statics.getResourceId(), statics.getPeriodType()
				.toString(), statics.getStatisticType().toString(),
				statics.getStartTime());
	}

	public static String createRowKey(String fqn, String period,
			String statisticType, long time) {
		StringBuilder strBuilder = new StringBuilder();
		strBuilder.append(fqn).append("_").append(period).append("_")
				.append(statisticType).append("_").append(time);
		return strBuilder.toString();
	}

}
